package cn.getech.data.development.utils.ws;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

@ServerEndpoint(value = "/ws/dataDevelopment/workflowLay/status/{workFlowMenuId}/{oozieId}")
@Component
@ConditionalOnProperty(name = "data-development.websocket-open", havingValue = "true")
public class WorkFlowLayStatusSocketServer {
    private static Logger log = LoggerFactory.getLogger(WorkFlowLayStatusSocketServer.class);
    private static final AtomicInteger onlineCount = new AtomicInteger(0);

    /***
     * concurrent包的线程安全Set，用来存放每个客户端对应的Session对象。
     * key: sessionId   value:Session
     */
    private static ConcurrentHashMap<String, Session> workflowLaySessionMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        log.info("websocket 加载");
    }

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("workFlowMenuId") String workFlowMenuId,@PathParam("oozieId")String oozieId) {
        String newKey = workFlowMenuId + "_" + oozieId + "_" + session.getId();
        workflowLaySessionMap.putIfAbsent(newKey, session);
        // 在线数加1
        int cnt = onlineCount.incrementAndGet();
        //log.info("有连接加入，当前连接数为：{}", cnt);
        sendMessage(session, "连接成功！");
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session, @PathParam("workFlowMenuId") String workFlowMenuId,@PathParam("oozieId")String oozieId) {
        //拼接对应的数据key: analysisId_type_sessionId   type:1 运行得到的实时数据  2：运行的实时日志
        String newKey = workFlowMenuId + "_" + oozieId + "_" + session.getId();
        workflowLaySessionMap.remove(newKey, session);
        if (session.isOpen()) {
            try {
                session.close();
            } catch (Exception e) {
                log.error("关闭连接异常：{}", e.getMessage());
            }
        }
        int cnt = onlineCount.decrementAndGet();
       // log.info("有连接关闭，当前连接数为：{}", cnt);
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        //log.info("来自客户端的消息：{}", message);
        sendMessage(session, "收到消息，消息内容：" + message);
    }

    /**
     * 出现错误
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误：{}，Session ID： {}", error.getMessage(), session.getId());
        error.printStackTrace();
    }

    /**
     * 发送消息，实践表明，每次浏览器刷新，session会发生变化。
     *
     * @param session
     * @param message
     */
    public static void sendMessage(Session session, String message) {
        try {
            session.getBasicRemote().sendText(message);
           // log.info("发送消息给id：{},消息为：{}", session.getId(), message);
        } catch (IOException e) {
            log.error("发送消息出错：{}", e.getMessage());
            e.printStackTrace();
        }
    }

    /**
     * 获取到全部的客户端用户
     */
    public static ConcurrentHashMap<String, Session> listSessionMap() {
        return workflowLaySessionMap;
    }
}
